package com.jcode.netty.client;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.jcode.netty.codec.NettyDecoder;
import com.jcode.netty.codec.NettyEncoder;
import com.jcode.netty.core.AbstractRemotingService;
import com.jcode.netty.core.ChannelWrapper;
import com.jcode.netty.process.DefaultRequestProcessor;
import com.jcode.netty.protocol.Packet;
import com.jcode.netty.serialize.Serializer;
import com.jcode.netty.serialize.SerializerFactory;
import com.jcode.netty.util.NamedThreadFactory;
import com.jcode.netty.util.RemotingHelper;
import com.jcode.netty.util.RemotingUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author dreajay
 * @DateTime 2016年12月13日 下午6:26:25
 */
public class NettyClient extends AbstractRemotingService {
    private Bootstrap bootstrap = new Bootstrap();
    private EventLoopGroup selectorWorkerEventLoopGroup;

    private volatile boolean start = false;

    private final ConcurrentHashMap<String, ChannelWrapper> channelTables
        = new ConcurrentHashMap<String, ChannelWrapper>();

    private final Lock lockChannelTables = new ReentrantLock();

    private ClientConfig config;

    private static final long LockTimeoutMillis = 50000;

    private static final Logger log = LoggerFactory.getLogger(NettyClient.class);

    public NettyClient(ClientConfig config) {
        super(config.getClientOnewaySemaphoreValue(), config.getClientAsyncSemaphoreValue());
        this.config = config;
        this.selectorWorkerEventLoopGroup = new NioEventLoopGroup(1,
            new NamedThreadFactory("selectorWorkerEventLoopGroup"));
        ExecutorService defaultRequestExecutor = Executors.newFixedThreadPool(config.getClientWorkerThreads(),
            new NamedThreadFactory("defaultRequestExecutor"));
        registerDefaultProcessor(new DefaultRequestProcessor(), defaultRequestExecutor);
        setCallbackExecutor(defaultRequestExecutor);
    }

    @Override
    public void start() {
        if (!start) {
            final Serializer serializer = SerializerFactory.getSerializer(config.getSerializeType());

            start = true;
            bootstrap.group(selectorWorkerEventLoopGroup).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .option(ChannelOption.SO_SNDBUF, config.getClientSocketSndBufSize())
                .option(ChannelOption.SO_RCVBUF, config.getClientSocketRcvBufSize())
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(
                            new NettyEncoder(serializer),
                            new NettyDecoder(),
                            new IdleStateHandler(0, 0, config.getClientChannelMaxIdleTimeSeconds()),//
                            new ClientHandler(NettyClient.this)
                        );
                    }
                });

            super.start();
            log.info("client start success...");
        } else {
            log.warn("client already started!");
        }

    }

    @Override
    public void shutdown() {
        if (start) {
            start = false;
            // 优雅退出，释放NIO线程组
            selectorWorkerEventLoopGroup.shutdownGracefully();
            super.shutdown();
        }
    }

    /**
     * 连接服务器
     *
     * @param address
     *
     * @return
     *
     * @throws InterruptedException
     */
    public void connectServer(final String address) throws Exception {
        getOrCreateChannel(address);
    }

    /**
     * 获取Channel，如果不存在，则重新创建
     *
     * @param address
     *
     * @return
     *
     * @throws InterruptedException
     */
    public Channel getChannel(final String address) throws Exception {
        ChannelWrapper cw = this.channelTables.get(address);
        if (cw != null && cw.isOK()) {
            return cw.getChannel();
        }
        return getOrCreateChannel(address);
    }

    public Channel getOrCreateChannel(final String address) throws Exception {
        ChannelWrapper cw = this.channelTables.get(address);
        if (cw != null && cw.isOK()) {
            return cw.getChannel();
        }
        if (this.lockChannelTables.tryLock(LockTimeoutMillis, TimeUnit.MILLISECONDS)) {
            try {
                boolean createNewConnection = false;
                cw = this.channelTables.get(address);
                if (cw != null) {
                    if (cw.isOK()) {
                        return cw.getChannel();
                    } else if (!cw.getChannelFuture().isDone()) {
                        createNewConnection = false;
                    } else {
                        this.channelTables.remove(address);
                        // TODO 是否需要关闭之前的
                        closeChannel(cw.getChannel());
                        createNewConnection = true;
                    }
                } else {
                    createNewConnection = true;
                }
                if (createNewConnection) {
                    ChannelFuture channelFuture = this.bootstrap.connect(RemotingHelper.string2SocketAddress(address));
                    log.info("createChannel: begin to connect remote host[{}] asynchronously", address);
                    cw = new ChannelWrapper(channelFuture);
                    this.channelTables.put(address, cw);
                }
            } catch (Exception e) {
                log.error("createChannel: create channel exception", e);
                throw e;
            } finally {
                this.lockChannelTables.unlock();
            }
        } else {
            log.warn("createChannel: try to lock channel table, but timeout, {}ms", LockTimeoutMillis);
        }
        if (cw != null) {
            ChannelFuture channelFuture = cw.getChannelFuture();
            if (channelFuture.awaitUninterruptibly(this.config.getConnectTimeoutMillis())) {
                if (cw.isOK()) {
                    log.info("createChannel: connect remote host[{}] success, {}", address, channelFuture.toString());
                    return cw.getChannel();
                } else {
                    log.warn("createChannel: connect remote host[" + address + "] failed, " + channelFuture.toString(),
                        channelFuture.cause());
                }
            } else {
                log.warn("createChannel: connect remote host[{}] timeout {}ms, {}", address,
                    this.config.getConnectTimeoutMillis(), channelFuture.toString());
            }
        }

        return null;
    }

    public void closeChannel(final Channel channel) throws Exception {
        if (null == channel) { return; }
        if (this.lockChannelTables.tryLock(LockTimeoutMillis, TimeUnit.MILLISECONDS)) {
            try {
                boolean removeItemFromTable = true;
                ChannelWrapper prevCW = null;
                String addrRemote = null;
                for (Map.Entry<String, ChannelWrapper> entry : channelTables.entrySet()) {
                    String key = entry.getKey();
                    ChannelWrapper prev = entry.getValue();
                    if (prev.getChannel() != null) {
                        if (prev.getChannel() == channel) {
                            prevCW = prev;
                            addrRemote = key;
                            break;
                        }
                    }
                }

                if (null == prevCW) {
                    log.info("eventCloseChannel: the channel[{}] has been removed from the channel table before",
                        addrRemote);
                    removeItemFromTable = false;
                }

                if (removeItemFromTable) {
                    this.channelTables.remove(addrRemote);
                    log.info("closeChannel: the channel[{}] was removed from channel table", addrRemote);
                    RemotingUtil.closeChannel(channel);
                }
            } catch (Exception e) {
                log.error("closeChannel: close the channel exception", e);
                throw e;
            } finally {
                this.lockChannelTables.unlock();
            }
        } else {
            log.warn("closeChannel: try to lock channel table, but timeout, {}ms", LockTimeoutMillis);
        }

    }

    public void sendRequestSync(Packet packet) throws Exception {
        sendRequestSync(getChannel(config.getServerAddress()), packet, config.getConnectTimeoutMillis());
    }

}
